package cn.edu.flink.tutorial.window;

import cn.edu.flink.tutorial.source.UDSourceFunction;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

public class TestWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Tuple2<Long, Long>> streamSource = env
                .addSource(new UDSourceFunction())
                .map(x -> Tuple2.of(x, 1L))
                .returns(Types.TUPLE(Types.LONG, Types.LONG));

        WatermarkStrategy<Tuple2<Long, Long>> watermarkStrategy = WatermarkStrategy
                .<Tuple2<Long, Long>>forMonotonousTimestamps()
                .withTimestampAssigner(new SerializableTimestampAssigner<Tuple2<Long, Long>>() {
                    @Override
                    public long extractTimestamp(Tuple2<Long, Long> element, long recordTimestamp) {
                        return element.f0;
                    }
                });

        SingleOutputStreamOperator<Tuple2<Long, Long>> streamData = streamSource.assignTimestampsAndWatermarks(watermarkStrategy);

        streamData
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
                .sum(1)
                .print();

        env.execute("TestWindow");
    }
}
